/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.fragmentation;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Properties;
import java.util.Vector;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.service.ServiceBulletin;
import cgl.narada.service.ServiceBulletinManager;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceNotification;
import cgl.narada.service.ServiceProperties;
import cgl.narada.service.Task;
import cgl.narada.service.TaskManager;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ProducerConstraints;
import cgl.narada.service.qos.impl.ProducerConstraintsImpl;
import cgl.narada.service.qos.impl.QosServiceImpl;

/** This class is responsible for fragmenting events. Based on the specified
    fragmentation constraints.
    
    @author Shrideep Pallickara
    $Date$
    $Revision$
*/

public class FragmentationService {
  private int DEFAULT_FRAGMENT_SIZE = 100000;
  private int DEFAULT_NUM_OF_FRAGMENTS = 5;
 
  private QosServiceImpl qosServiceImpl;
  private NBEventGenerator nbEventGenerator;
  
  private String fragDirectory = "C:/TempFiles/tmpFiles/frag";
  
  private String fragExt = ".nbf";
  private String fragInfoExt = ".nbi";
  private String metaExt = ".nbm";
  

  private Hashtable metadataTable;
  private Hashtable taskTable;
  private ProducerConstraints fragmentConstraints;
  private boolean debug = false;
  private String moduleName = "FragmentationService: ";
  
  /** This is to be used for testing purposes only */
  protected FragmentationService() {
    this.nbEventGenerator = new NBEventGenerator();
    metadataTable = new Hashtable();
  }
  
  public FragmentationService(QosServiceImpl qosServiceImpl) {
    this.qosServiceImpl = qosServiceImpl;
    
    ServiceProperties serviceProperties = qosServiceImpl.getServiceProperties();
    String dirValue = serviceProperties.getProperty("FragmentationDirectory");
    if (dirValue != null) {
      fragDirectory = dirValue;
      if (debug) {
          System.out.println(moduleName + "Fragmentation Directory value = " + 
                             fragDirectory);
      }
    } 
    
    this.nbEventGenerator = qosServiceImpl.getEventGenerator();
    metadataTable = new Hashtable();
    taskTable = new Hashtable();
  }
  
  public synchronized void 
  fragmentEvent(NBEvent nbEvent, ProducerConstraints producerConstraints)  
    throws ServiceException {
    int fragmentSize = 0;
    int numOfFragments = 0;
    byte[] contentPayload = nbEvent.getContentPayload();
    Properties fragmentationProperties = 
      producerConstraints.getFragmentationParams();
    
    /** This needs to be cleaner. Fix this. **********************/
    int _entityId = producerConstraints.getEntityId();
    int _producerId = 
      ((ProducerConstraintsImpl) producerConstraints).getProducerId();
    TemplateInfo _templateInfo =
      producerConstraints.getTemplateInfo();
    fragmentConstraints = new ProducerConstraintsImpl(_entityId, _producerId,
						      _templateInfo);
    /*************************************************************/

    
    String fragmentSizeString = 
      fragmentationProperties.getProperty("fragmentSize"); 

    if (fragmentSizeString != null) {
      fragmentSize = Integer.parseInt(fragmentSizeString);
    }

    String numOfFragmentsString = 
      fragmentationProperties.getProperty("numberOfFragments"); 

    if (numOfFragmentsString != null) {
      numOfFragments = Integer.parseInt(numOfFragmentsString);
    }

    if (fragmentSize == 0 && numOfFragments == 0) {
      throw new ServiceException(moduleName + "Neither [fragmentSize] OR [" +
				 "numberOfFragments] specified correctly");
    }
    
    if (fragmentSize != 0 && numOfFragments != 0) {
      throw new ServiceException(moduleName + "Simultaneos specification of ["+
				 "fragmentSize] AND [numberOfFragments] is " +
				 "NOT allowed");
    }
    
    String largeFileLoc = 
      fragmentationProperties.getProperty("fileLocation");
    if (largeFileLoc == null) {
      throw new ServiceException(moduleName + "The property fileLocation has "
				 + "not been specified ");		      
    }
    
    File largeFile = new File(largeFileLoc);
    
    if (  (!largeFile.exists()) || (!largeFile.isFile()) ) {
      throw new ServiceException(moduleName + "The specified file [" + 
				 largeFileLoc + 
				 "] does not exist. Please check " +
				 "specified file OR see if its a directory!");
    }
    
    long contentLength = largeFile.length();
    System.out.println(moduleName + "The specified file [" + largeFileLoc + 
		       "] is of length = " + contentLength);
    
    if (contentLength == 0) {
      throw new ServiceException(moduleName + "The file [" + largeFileLoc + 
				 "] has a contentLength of ZERO!");
    }
    
    /** Now that we have reached here, only one of either fragmentSize or
	numOfFragments WILL be non-zero */
    boolean shouldFragment = true;
    int remaining=0;
    int lastFragmentSize=0;
    
    if (fragmentSize != 0) {
      if (contentLength <= fragmentSize) {
	System.out.println(moduleName + "No fragmentation necessary  since " +
			   "contentLength=" + contentLength + 
			   " <= fragmentSize=" + fragmentSize);
	shouldFragment = false;
      } else {
	numOfFragments = (int) (contentLength/(long)fragmentSize );
	lastFragmentSize = fragmentSize;
	remaining = (int) ( contentLength - 
			    (long)(numOfFragments*fragmentSize) );
	if ( remaining != 0) {
	  numOfFragments++;
	  lastFragmentSize =  remaining;
	  System.out.println(moduleName + "LAST FRAGMENT SIZE = " + 
			     lastFragmentSize);
	}
      }
      
    } else {
      if (contentLength <= numOfFragments) {
	System.out.println(moduleName + "No fragmentation necessary since " +
			   "contentLength=" + contentLength + 
			   " <= numOfFragment" + numOfFragments);
	shouldFragment = false;
      } else {
	fragmentSize = (int) (contentLength/(long)numOfFragments );
	remaining = (int) ( contentLength - 
			    (long)(numOfFragments*fragmentSize) );
	lastFragmentSize = fragmentSize + remaining;
      }
    }
    
    
    
    if (shouldFragment) {
      System.out.println(moduleName + "Beginning fragmentation process!");
      manageFileFragmentation(nbEvent, largeFileLoc, numOfFragments, 
			      fragmentSize,  lastFragmentSize);
    }
    
    System.out.println(moduleName + "Fragmentation process complete ..");
  }
  
  
  
  private void 
  manageFileFragmentation(NBEvent nbEvent, String largeFileLoc, 
			  int numOfFragments, int fragmentSize, 
			  int lastFragmentSize) throws ServiceException {
    File largeFile = new File(largeFileLoc);
    String filename = largeFile.getName();
    String fragDirLocation = fragDirectory + "/" + filename + "-tempDir";
    long contentLength = largeFile.length();

    initializeTask(filename, numOfFragments);

    /** First create the fragments, and store it. These fragments are
	what would be used for retransmissions and such */
    createAndStoreFragments(fragDirLocation, largeFileLoc, 
			    numOfFragments, fragmentSize, lastFragmentSize);
    
    /** Next create the metadata associated with the robust delivery of this
	file */
    FileFragmentationMetaData ffmd = 
      createMetadata(nbEvent, fragDirLocation, largeFileLoc, contentLength, 
		     numOfFragments, fragmentSize, lastFragmentSize);
    metadataTable.put(filename, ffmd);

    /** Retrieve individual fragments and try to publish them */
    retrieveAndPublishFragments(nbEvent, filename, contentLength, 
				fragDirLocation, numOfFragments);
  }
  
  
  private void 
  createAndStoreFragments(String fragDirLocation, String largeFileLoc, 
			  int numOfFragments, int fragmentSize, 
			  int lastFragmentSize) 
    throws ServiceException {
    try {
      
     
      System.out.println(moduleName + "Will create and store fragments to [" +
			 fragDirLocation + "]");
      
      File largeFile = new File(largeFileLoc);
      String filename = largeFile.getName();
      FileInputStream fis = new FileInputStream(largeFile);
      long totalWritten=0;
      
      //ProgressIndicator pi = new ProgressIndicator();
      //long currCompletion = 0;
      //long prevCompletion = 0;
      int bytesRead = 0;
      for (int i = 1; i < numOfFragments; i++) {
	controlMemoryUtilization();
	byte[] fragment = new byte[fragmentSize];
	bytesRead = fis.read(fragment);
	
	if (bytesRead != fragmentSize) {
	  System.out.println(moduleName + "Tried to read (" + fragmentSize + 
			     ") bytes, ended up reading (" + bytesRead + 
			     ") bytes!");
	}

	writeIndividualFragment(fragDirLocation, i , fragment);
	totalWritten += fragmentSize;
	reflectTaskProgression(filename, 0, i);
	//double percentComplete =   
	//  Math.floor ( (  ((double)i/(double)numOfFragments) * 100 ) );
	
	//currCompletion = (long) percentComplete;
	//pi.showUpdate(currCompletion, prevCompletion, 
	//	      " % of fragments created");
	//prevCompletion = currCompletion;
      }
      
      /** Proceed to write the last fragment */
      byte[] fragment = new byte[lastFragmentSize];
      bytesRead = fis.read(fragment);
      if (bytesRead != fragmentSize) {
	System.out.println(moduleName + "Tried to read (" + fragmentSize + 
			   ") bytes, ended up reading (" + bytesRead + 
			   ") bytes!");
      }
      
      writeIndividualFragment(fragDirLocation, numOfFragments, fragment);
      totalWritten += lastFragmentSize;

      reflectTaskProgression(filename, 0, numOfFragments);

      fis.close();
      System.out.println(moduleName + "Total bytes written = " + totalWritten);

    } catch (IOException ioe) {
      throw new ServiceException(moduleName + 
				 "Error in createAndStoreFragments()->" + ioe);
    }
  }

  /** This method is used to write an individual fragment to the specified
      location. */
  private void writeIndividualFragment(String fragDirLocation, int fragmentNum,
				      byte[] fragment) 
    throws ServiceException {
    try {
      /** Make sure that the temporary fragmentation directory exists. If it
	  does not proceed to create the directory */
      File fragDirectory = new File(fragDirLocation);
      if (!fragDirectory.exists()) {
	fragDirectory.mkdirs();
      }
      
      /** Create the location of the fragmentation file that we would be
	  writing the fragment to */
      String fragFileName = 
	fragDirLocation + "/fragment" + fragmentNum + fragExt;
      File fragFile = new File(fragFileName);
      fragFile.createNewFile();
      FileOutputStream fos = new FileOutputStream(fragFile);
      fos.write(fragment);
      fos.close();
    } catch (IOException ioe) {
      throw new ServiceException(moduleName + 
				 "Error in writeIndividualFragment()->" + ioe);
    }
  }
  
  


  private FileFragmentationMetaData 
  createMetadata(NBEvent nbEvent, String fragDirLocation, String largeFileLoc, 
		 long contentLength, 
		 int numOfFragments, int fragmentSize, int lastFragmentSize)
    throws ServiceException {
    
    try {
      File largeFile = new File(largeFileLoc);
      String metaFileLoc = 
	fragDirLocation + "/" + largeFile.getName() + metaExt;
      
      FileFragmentationMetaData ffmd = 
	new FileFragmentationMetaData(nbEvent, fragDirLocation, largeFileLoc, 
				      contentLength, numOfFragments,
				      fragmentSize, lastFragmentSize,
				      System.currentTimeMillis(),
				      fragmentConstraints.getTemplateInfo(),
				      fragmentConstraints.getEntityId());
      
      File metaFile = new File(metaFileLoc);
      FileOutputStream fos = new FileOutputStream(metaFile);
      System.out.println(ffmd);
      byte[] metadata = ffmd.getBytes();
      fos.write(metadata);
      fos.close();
      
      return ffmd;

    } catch (IOException ioe) {
      throw new ServiceException(moduleName + "Error in createMetadata()-> " +
				 ioe);
    }
  }


  
  private void 
  retrieveAndPublishFragments(NBEvent nbEvent, String largeFilename,
			      long contentLength, 
			      String fragDirLocation, int numOfFragments) 
    throws ServiceException {
    
    for (int i=1; i < numOfFragments +1; i++) {
      publishIndividualFragment(nbEvent, largeFilename, contentLength,
				fragDirLocation, i, numOfFragments);
      reflectTaskProgression(largeFilename, 1, i);
    }
    
  }
  
  
  private void 
  publishIndividualFragment(NBEvent nbEvent, String largeFilename,
			    long contentLength, String fragDirLocation, 
			    int fragmentNumber, int numOfFragments) 
    throws ServiceException {
    controlMemoryUtilization();
    FileFragmentationMetaData ffmd =
      (FileFragmentationMetaData) metadataTable.get(largeFilename);
    long fragmentationEpoch = ffmd.getTimestamp();
    
    byte[] fragmentBytes = 
      retrieveIndividualFragment(fragDirLocation, fragmentNumber);
    NBEvent fragmentEvent = 
      nbEventGenerator.getFragmentedEvent(nbEvent, largeFilename, 
					  contentLength, fragmentBytes, 
					  fragmentNumber, numOfFragments,
					  fragmentationEpoch);
    if (qosServiceImpl != null) {
      qosServiceImpl.publishEvent(fragmentEvent, fragmentConstraints, 2000);
    }
   
  }
  


  
  private byte[] 
  retrieveIndividualFragment(String fragDirLocation, int fragmentNumber) 
    throws ServiceException {
    try {
      /** Create the location of the fragmentation file that we would be
	  writing the fragment to */
      String fragFileName = 
	fragDirLocation + "/fragment" + fragmentNumber + fragExt;
      
      File fragmentFile = new File(fragFileName);
      if (!fragmentFile.exists()) {
	throw new ServiceException(moduleName + "Fragment file " + fragFileName
				   + " NOT found");
      }

      int contentLength = (int) fragmentFile.length();
      if (contentLength == 0) {
	throw new ServiceException(moduleName + "Fragment file " + fragFileName
				   + " contentLength=0");
      }
      
      byte[] fragmentBytes = new byte[contentLength];
      
      FileInputStream fis = new FileInputStream(fragmentFile);
      int bytesRead = fis.read(fragmentBytes);
      fis.close();
      if (bytesRead != contentLength) {
	System.out.println(moduleName + "Only (" + bytesRead + 
			   ") bytes of Fragment file with length=" +
			   contentLength + " were read");
	
	throw new ServiceException(moduleName + "Only (" + bytesRead + 
				   ") bytes of Fragment file with length=" +
				   contentLength + " were read");
      }
      
      return fragmentBytes;

    } catch (IOException ioe) {
      throw new ServiceException(moduleName + 
				 "Error in retrieveIndividualFragment()-> " +
				 ioe);
    }
  }
  

  /** This is used by the Reliable delivery algorithm to satisfy requests
      for retransmissions */  
  public byte[] 
  retrievePayloadForFragment(String filename, int fragmentNumber) 
    throws ServiceException {
    if (!metadataTable.containsKey(filename)) {
      throw new ServiceException(moduleName + "Metadata for [" + filename + 
				 "] is not available");
    }
    
    FileFragmentationMetaData ffmd = 
      (FileFragmentationMetaData) metadataTable.get(filename);
    
    String fragDirLoc = ffmd.getFragmentationDirectory();
    return retrieveIndividualFragment(fragDirLoc, fragmentNumber);
  }

  
  /** This is used by the reliable delivery algorithm to update information
      regarding the catenation number associated with the fragment for a 
      specific file */
  public void assignedCatenationNum(String filename, int fragmentNum, 
				    int catenationNum) 
    throws ServiceException  {
    if (!metadataTable.containsKey(filename)) {
      throw new ServiceException(moduleName + "Metadata for [" + filename + 
				 "] is not available");
    }
    
    FileFragmentationMetaData ffmd = 
      (FileFragmentationMetaData) metadataTable.get(filename);
    
    String fragDirLoc = ffmd.getFragmentationDirectory();
    File fragDir = new File(fragDirLoc);
    if (!fragDir.exists()) {
      throw new ServiceException(moduleName + "The directory [" + fragDir + 
				 "] should have already been "
				 + " created during the meta-data process");
    }
    
    String fragInfoLoc = fragDirLoc + "/Fragment" + fragmentNum + fragInfoExt;
    
    
    try {
      File fragmentInfoFile = new File(fragInfoLoc);
      fragmentInfoFile.createNewFile();
      //System.out.println(moduleName + "Writing file " + 
      //fragmentInfoFile.getAbsolutePath());
      
      FragmentCatenationInfo fragCatInfo = 
	new FragmentCatenationInfo(filename, fragmentNum, catenationNum);
      //System.out.println(fragCatInfo);
      FileOutputStream fos = new FileOutputStream(fragmentInfoFile);
      byte[] fragdata = fragCatInfo.getBytes();
      fos.write(fragdata);
      fos.close();
    } catch (IOException ioe) {
      throw new ServiceException(moduleName + 
				 "Error in assignedCatenationNum()->" + ioe);
    }
  }
  

  /** This is used by the reliable delivery service to manage the successful
      storage of the fragment of a file */
  public void processReliableStorageOf(String filename, int fragmentNumber) 
    throws ServiceException {
     if (!metadataTable.containsKey(filename)) {
      throw new ServiceException(moduleName + "Metadata for [" + filename + 
				 "] is not available");
    }
    
    FileFragmentationMetaData ffmd = 
      (FileFragmentationMetaData) metadataTable.get(filename);
    
    String fragDirLocation = ffmd.getFragmentationDirectory();
    File fragDir = new File(fragDirLocation);
    if (!fragDir.exists()) {
      throw new ServiceException(moduleName + "The directory [" + 
				 fragDirLocation + "] should have already been"
				 + " created during the meta-data process");
    }
    
    String fragFileName = 
      fragDirLocation + "/fragment" + fragmentNumber + fragExt;
    
    String fragInfoLoc = 
      fragDirLocation + "/Fragment" + fragmentNumber + fragInfoExt;
    
    File fragmentFile = new File(fragFileName);
    File fragInfoFile = new File(fragInfoLoc);

    if (!fragInfoFile.isFile() || !fragInfoFile.exists()) {
      System.out.println(moduleName + "FragmentInfo file [" + fragInfoLoc +
			 "] not available!");
      return;
    }

    
    if (fragInfoFile.delete() ) {
      System.out.println(moduleName + "FragmentInfo file [" + fragInfoFile + 
			 "] deleted!");
    } else {
      System.out.println(moduleName + "FragmentInfo file [" + fragInfoFile + 
			 "] NOT DELETED!");
    }
    
    reflectTaskProgression(filename, 2, fragmentNumber);
    if (fragmentNumber == ffmd.getNumberOfFragments()) {
      processCompletionOfFragmentedDelivery(ffmd);
    }

  }
  
  /** This method is called once we are assured that all the fragments that
      needed to be delivered have been delivered. */
  private void 
  processCompletionOfFragmentedDelivery(FileFragmentationMetaData ffmd) 
    throws ServiceException {
    String fragDirLocation = ffmd.getFragmentationDirectory();
    String metaFileLoc = 
      fragDirLocation + "/" + ffmd.getFileName() + metaExt;
    
    File metaFile = new File(metaFileLoc);
    
    if (!metaFile.exists()) {
      System.out.println(moduleName + "The metafile [" + metaFileLoc +
			 "] is not available. Will not process further!");
      return;
    }
    
    metaFile.delete();
    int numOfFragments = ffmd.getNumberOfFragments();


    for (int i=1; i <numOfFragments + 1; i++) {
      String fragFileName = 
	fragDirLocation + "/fragment" + i + fragExt;
      File fragmentFile = new File(fragFileName);
      if (!fragmentFile.exists()) {
	System.out.println(moduleName + "The fragment file [" + fragFileName +
			   "] is not available. Will not process further!");
	return;
      }
    }

    File fragmentDirectory = new File(fragDirLocation);

    if (!fragmentDirectory.exists()) {
      System.out.println(moduleName + "The fragment DIRECTORY [" + 
			 fragDirLocation  +
			 "] is not available. Will not process further!");
      return;
    }
    
    if (!fragDirLocation.endsWith("-tempDir")) {
      System.out.println(moduleName + "The fragment DIRECTORY [" + 
			 fragDirLocation + "] does not end with [-tempDir]"
			 + ".Will not process further!");
      return; 
    }
    
    System.out.println("\n\n\n" + moduleName + "Scheduling fragmentation " +
		       "directory [" + fragDirLocation + "] for deletion!");
    deleteFragmentationDirectory(fragDirLocation);
  }

  /** This method should be called only after the check have been made to
      make sure that this is a temporary fragmentation directory */
  private void deleteFragmentationDirectory(String fragDirLocation) {
    File fragmentDirectory = new File(fragDirLocation);

    if (!fragmentDirectory.exists()) {
      System.out.println(moduleName + "The fragment DIRECTORY [" + 
			 fragDirLocation  +
			 "] is not available. Will not process further!");
      return;
    }

    if (!fragmentDirectory.isDirectory()) {
      System.out.println(moduleName + "The directory " + fragDirLocation + 
			 " that was supplied is not available. Returning!"); 
    }
    
    if (!fragDirLocation.endsWith("-tempDir")) {
      System.out.println(moduleName + "The fragment DIRECTORY [" + 
			 fragDirLocation + "] does not end with [-tempDir]"
			 + ". Will not process further!");
      return; 
    }
    
    File[] fileList = fragmentDirectory.listFiles();
    Vector couldNotDelete = null;
    
    boolean containsNonNBFiles = false;
    for (int i=0; i < fileList.length; i++) {
      if ( fileCanBeDeleted(fileList[i]) ) {
	fileList[i].delete();
      } else {
	containsNonNBFiles = true;
	if (couldNotDelete == null) {
	  couldNotDelete = new Vector();
	}
	couldNotDelete.addElement(fileList[i].getName());
      }/* end if */
    }/* end for */
    
    if (containsNonNBFiles) {
      System.out.println(moduleName + "The directory [" + fragDirLocation +
			 "] contains (" + couldNotDelete.size() + ") files =>"
			 + couldNotDelete + " that were not created by NB." + 
			 "Directory WILL NOT BE Deleted");
      return;
    }

    fragmentDirectory.delete();
  }
  
  /** We need to be absolutely sure that the file we delete is a file that we
      originally created. We do not allow the deletion of any file that does
      not have the extensions which files created by NB have. */  
  private boolean fileCanBeDeleted(File file) {
    String filename = file.getName();
    if (filename.endsWith(fragExt)) {
      return true;
    }
    
    if (filename.endsWith(fragInfoExt)) {
      return true;
    }
    
    if (filename.endsWith(metaExt)) {
      return true;
    }

    return false;
  }

  public void checkForPendingTransfers(int publisherCatenationNumber) 
    throws ServiceException {
    File file = new File(fragDirectory);
    if (!file.isDirectory()) {
      System.out.println(moduleName + file + " is NOT a directory. ");
      return;
    }
    File[] pendingFiles =  file.listFiles();
    
    Hashtable pendingMetadataTable = new Hashtable();
    for (int i=0; i < pendingFiles.length; i++) {
      if (pendingFiles[i].isDirectory()) {
	System.out.println(moduleName + "Processing directory " + 
			   pendingFiles[i]);
	//processRepublishingFor(pendingFiles[i]);
	FileFragmentationMetaData _metadata = 
	  retrieveMetadataFor(pendingFiles[i]);
	if (_metadata != null) {
	  pendingMetadataTable.put(new Long(_metadata.getTimestamp()),
				   _metadata);
	  metadataTable.put(_metadata.getFileName(), _metadata);
	}/* metadata != null*/
	

      } /* end if */
    }/*end for */

    
    if (pendingMetadataTable.size() != 0) {
      long[] timestamps = new long[pendingMetadataTable.size()];
      
      Enumeration e = pendingMetadataTable.keys();
      int tracker=0;
      while (e.hasMoreElements()) {
	timestamps[tracker++] = ((Long)e.nextElement()).longValue();
      }/*end while */

      Arrays.sort(timestamps);

      for (int i=0; i< timestamps.length; i++) {
	Long timeKey = new Long(timestamps[i]);
	FileFragmentationMetaData ffmd = 
	  (FileFragmentationMetaData) pendingMetadataTable.get(timeKey);
	processRepublishingFor(ffmd, publisherCatenationNumber);
      }
    }
  }




  private FileFragmentationMetaData retrieveMetadataFor(File directoryFile) {
    String directoryName = directoryFile.getName();
    int subIndex = new String("-tempDir").length();
    String pendingFileString = 
      directoryName.substring(0, directoryName.length() - subIndex);
    String metaFileLoc = directoryFile.getAbsolutePath() + 
      "/" + pendingFileString + metaExt;
    
    File metadataFile = new File(metaFileLoc);
    
    try {
      FileInputStream fis = new FileInputStream(metadataFile);
      byte[] metadata = new byte[(int)metadataFile.length()];
      fis.read(metadata);
      fis.close();
      FileFragmentationMetaData ffmd = new FileFragmentationMetaData(metadata);
      
      return ffmd;
    } catch (IOException ioe) {
      System.out.println(moduleName + ioe);
    }
    return null;
  }
  
  
  private void 
  processRepublishingFor(FileFragmentationMetaData ffmd,
			 int publisherCatenationNumber) 
    throws ServiceException {
    System.out.println(moduleName + "Will process republishing for " + ffmd);
    
    int totalNumOfFragments = ffmd.getNumberOfFragments();
    String fragDirLoc = ffmd.getFragmentationDirectory();
    String filename = ffmd.getFileName();
    
    NBEvent nbEvent = ffmd.getNBEvent();
    long contentLength = ffmd.getContentLength();
    
    
    boolean shouldBeTrueFromNow = false;
    TemplateInfo templateInfo = ffmd.getTemplateInfo();
    int entityId = ffmd.getEntityId();
    
    initializeTask(filename, totalNumOfFragments);
    reflectTaskProgression(filename, 0, totalNumOfFragments);
    reflectTaskProgression(filename, 1, publisherCatenationNumber);

    fragmentConstraints = 
      new ProducerConstraintsImpl(entityId, 2000, templateInfo);
    
    for (int i=1; i < totalNumOfFragments+1; i++) {
      String fragInfoLoc = fragDirLoc + "/Fragment" + i + fragInfoExt; 
      File fragInfoFile = new File(fragInfoLoc);
      
      if (fragInfoFile.exists()) {
	//shouldBeTrueFromNow = true;
	byte[] fgInfoBytes = new byte[(int)fragInfoFile.length()];
	try {
	  FileInputStream fis = new FileInputStream(fragInfoFile);
	  fis.read(fgInfoBytes);
	  fis.close();
	} catch (IOException ioe) {
	  System.out.println(moduleName + "Error reading from ["+fragInfoLoc + 
			     "] " + ioe );
	  return;
	}
	FragmentCatenationInfo fragCatInfo = 
	  new FragmentCatenationInfo(fgInfoBytes);
	int _fragCat = fragCatInfo.getCatenationNumber();
	if ( _fragCat <= publisherCatenationNumber && 
	     _fragCat !=0 ) {
	  System.out.println(moduleName + "Based on publisher catenation (" +
			     publisherCatenationNumber + ") fragment (" + i + 
			     ") with catenation (" + _fragCat + 
			     ") was reliably stored ...");
	  processReliableStorageOf(filename, i);
	} else {
	  System.out.println(moduleName + "Will proceed with republishing " +
			     "fragment (" + i + ") for file [" + filename 
			     + "]");
	  
	  publishIndividualFragment(nbEvent, filename, contentLength, 
				    fragDirLoc, i, totalNumOfFragments);
	  reflectTaskProgression(filename, 1, i);
	}
      } else {
	
	/*if (shouldBeTrueFromNow) {
	  System.out.println("\n\n" + moduleName + "The file [" + fragInfoLoc 
	  + "] should have been present. CANNOT PROCEED" +
	  " FURTHER. \n\n\n");
	  return;
	  }*/
	System.out.println(moduleName + "Will proceed with republishing " +
			   "fragment (" + i + ") for file [" + filename 
			   + "]");
	
	publishIndividualFragment(nbEvent, filename, contentLength, 
				  fragDirLoc, i, totalNumOfFragments);
	
	reflectTaskProgression(filename, 1, i);
      } /* end fragInfoFile.exists() */
    } /*end for totalNumOfFragments */
  }



  private void controlMemoryUtilization() {
    long totalMemory=0;
    long UTIL_LIMIT = 45000000;

    totalMemory = Runtime.getRuntime().totalMemory();
    
    while (totalMemory > UTIL_LIMIT) {
      Thread _currentThread = Thread.currentThread();
      System.out.println(moduleName + "Will yield()->" + _currentThread + 
			 " to control Memory Utilization of [" + 
			 totalMemory*0.000001 + "] MBytes");
      Runtime.getRuntime().gc();
      _currentThread.yield();
      
      totalMemory = Runtime.getRuntime().totalMemory();
    }/* end while */

  }

  
  public static void main(String[] args) {
    FragmentationService fragmentor = new FragmentationService();
    
    String testString = "The Community Grids Lab is located at IU";

    byte[] testBytes = testString.getBytes();
    
    String filename = "D:/TestFiles/tester.txt";
    
    try {
      ClientService clientService = SessionService.getClientService(7007);
      EventProducer producer = clientService.createEventProducer();
      TemplateInfo templateInfo = 
	clientService.createTemplateInfo(12345, 
			       TemplateProfileAndSynopsisTypes.STRING,
					 "Movie/Casablanca");
      ProducerConstraints producerConstraints = 
	producer.createProducerConstraints(templateInfo);
      producer.generateEventIdentifier(true);
      producer.setTemplateId(12345);
      producer.setDisableTimestamp(false);
      String stringMsg = "testing fragmentation!";
      NBEvent nbEvent = 
	producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
			       "Movie/Casablanca",
			       stringMsg.getBytes());
      Properties fragmentationProperties =  new Properties();
      fragmentationProperties.put("fragmentSize", "250");
      fragmentationProperties.put("fileLocation", filename);
      
      producerConstraints.setSendAfterFragmentation(fragmentationProperties);
      fragmentor.fragmentEvent(nbEvent, producerConstraints);
    
      int numOfFragments = 88;
      int catenationStart = 1001;

      for (int i=1; i < numOfFragments +1; i++) {
	fragmentor.assignedCatenationNum("tester.txt", i, catenationStart++); 
      }
      
      
      for (int i=1; i < numOfFragments - 20; i++) {
	//fragmentor.processReliableStorageOf("tester.txt", i); 
      }

      fragmentor.checkForPendingTransfers(catenationStart - 20);

    } catch (ServiceException serEx) {
      System.out.println(serEx);
    }

  } 


  private void initializeTask(String filename, int numOfFragments) {
    if (taskTable.containsKey(filename)) {
      System.out.println(moduleName + "Removing task info pertaining to a" +
			 " previous file transfer");
      taskTable.remove(filename);
    }

    Task task = TaskManager.createTask();
    TaskManager.setTaskDescription(task, "The robust transfer of file [" +
				   filename + "]");
    TaskManager.createSubTasks(task, 3);
    
    Task subTask0 = task.getSubTask(0);
    TaskManager.setTaskDescription(subTask0, "Creating file fragments");
    TaskManager.setNumOfWorkUnits(subTask0, numOfFragments);
    
    
    Task subTask1 = task.getSubTask(1);
    TaskManager.setTaskDescription(subTask1, "Publishing File fragments");
    TaskManager.setNumOfWorkUnits(subTask1, numOfFragments);


    Task subTask2 = task.getSubTask(2);
    TaskManager.setTaskDescription(subTask2, 
				   "Ensured reliable delivery of fragments");
    TaskManager.setNumOfWorkUnits(subTask2, numOfFragments);

    taskTable.put(filename, task);
  }
  
  private void reflectTaskProgression(String filename, int subTaskNum,
				      int numOfWorkUnitsCompleted) {
    Task task = (Task) taskTable.get(filename);

    if (task == null) {
      System.out.println(moduleName + "Trying to update task progression on" +
			 " an unknown task!");
      return;
    }

    Task subTask = task.getSubTask(subTaskNum);
    TaskManager.setNumOfWorkUnitsCompleted(subTask, numOfWorkUnitsCompleted);

    int entityId = qosServiceImpl.getEntityId();
    issueServiceNotification(entityId, task);
  }
  
  
  private void issueServiceNotification(int entityId, Task taskUpdate) {
    ServiceBulletin bulletin = 
      ServiceBulletinManager.getServiceBulletin(entityId);
    
    if (bulletin == null) {
      System.out.println(moduleName + "No ServiceBulletin registered for " +
			 "entity [" + entityId + "]");
      return;
    }
    
    ServiceNotification notification = new ServiceNotification();
    notification.addTaskInfo(taskUpdate);
    bulletin.onServiceNotification(notification);
  }
  
}









